package com.houshuai.emqtt.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.mqtt.core.ConsumerStopAction;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.router.PayloadTypeRouter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * Spring Integration基于Eclipse Paho MQTT客户端库提供了支持MQTT协议的入站和出站通道适配器。
 * <p>
 * FileName: RedisConfig
 * author:   候帅
 * data:     2018/7/9 21:10
 * Description: TODO
 * History:
 * <p>
 * 3.2 配置MQTT订阅者
 * Inbound 入站消息适配器
 * <p>
 * 第一步：配置MQTT客户端工厂类DefaultMqttPahoClientFactory
 * <p>
 * 第二步：配置MQTT入站消息适配器MqttPahoMessageDrivenChannelAdapter
 * <p>
 * 第三步：定义MQTT入站消息通道MessageChannel
 * <p>
 * 第四步：声明MQTT入站消息处理器MessageHandler
 * <p>
 * 3.3 配置MQTT发布者
 * Outbound 出站消息适配器
 * <p>
 * 第一步：配置Outbound出站，出站通道适配器
 * <p>
 * 第二步：配置Outbound出站，发布者发送的消息通道
 * <p>
 * 第三步：对外提供推送消息的接口
 */
@Slf4j
@Configuration
public class MqttConfiguration {

    @Autowired
    private MqttProperties mqttProperties;

    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(mqttProperties.getUsername());
        mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{mqttProperties.getHostUrl()});
        mqttConnectOptions.setKeepAliveInterval(2);
        return mqttConnectOptions;
    }

    /**
     * 客户端工厂类,根据配置的选项(用户名、密码、服务器集群地址等)创建一个默认的客户端。
     * 配置DefaultMqttPahoClientFactory
     * 1. 配置基本的链接信息
     * 2. 配置maxInflight，在mqtt消息量比较大的情况下将值设大
     *
     * @return
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
//        factory.getAsyncClientInstance(mqttProperties.getHostUrl(),mqttProperties.getClientId()).setCallback();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }


    /**
     * MQTT出站通道适配器的抽象类的实现,用于推送消息。
     * 服务激活器
     * 当从入站通道接收到消息时，这允许我们在POJO上执行任何方法，并且它允许我们将消息写入一个外向通道。
     * <p>
     * Service Activator可调用Spring的Bean来处理消息，并将处理后的结果输出到指定的消息通道。
     * 服务激活器从输入信道接收消息并发送这些消息给的 MessageHandler
     * MessageHandler 实现开箱即用（PayloadTypeRouter 就是 MessageHandler 的实现）
     * <p>
     * inputChannel参数指定了用于接收消息信息的channel
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "pubSubChannel")//mqttOutboundChannel
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getClientId(), mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(0);
//        messageHandler.setCompletionTimeout(1000);
        messageHandler.setDefaultTopic(mqttProperties.getDefaultTopic());

        return messageHandler;


    }

    /**
     * 输入通道
     *
     * @return DirectChannel的关键特性是，尽管它在发布-订阅模型上工作，但它以循环方式将它接收到的每条消息只提供给其中一个订阅者。
     * 因此，它是发布-订阅和点对点传输模型的组合。
     * <p>
     * DirectChannel是Spring Integration默认的消息通道，它允许将消息发送给一个订阅者，然后阻碍发送直到消息被接收
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }


    /**
     * 允许广播消息给所有订阅者
     *
     * @return
     */
    @Bean(name = "pubSubChannel")
    public MessageChannel pubSubChannel() {
        PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel();
        return publishSubscribeChannel;
    }

}
